package com.wzl.consumer.controller.kafkaController.basicsConsumer;

import com.alibaba.fastjson.JSONObject;
import com.wzl.common.VkPublicResponse;
import com.wzl.common.errcode.VkErrorCodeEnum;
import com.wzl.common.exception.VkPublicException;
import com.wzl.common.pojo.VkSmbmsUser;
import com.wzl.consumer.service.kafkaService.basicsConsumer.MqBasicsKafkaConService;
import com.wzl.producer.config.kafkaConfig.KafkaConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.List;

/**
 * kafka消费者 控制器
 *
 * @author weizhongli
 * @since 2022-09-13 16:47
 */
@RestController
@Slf4j
public class MqBasicsKafkaConController {

    @Resource
    private MqBasicsKafkaConService mqBasicsKafkaConService;


    /**
     * 基本消费监听
     */
    //@KafkaListener(topics = {KafkaConfig.TOPIC}, id = "consumer", containerGroup = "${spring.kafka.consumer.group-id1}")
    public VkPublicResponse onMessage1(ConsumerRecord<?, ?> record, Acknowledgment ack){
        // 消费的哪个topic、partition的消息,打印出消息内容
        log.info("consumer开始消费--->topic名字:"+record.topic()+",分区partition:"+record.partition()+",消费数据"+record.value());
        //解析收到的队列内容
        Object value = record.value();
        JSONObject jsonMsg = JSONObject.parseObject(value.toString());
        VkSmbmsUser user = createUssCrmEntity(jsonMsg);
        mqBasicsKafkaConService.updateUserById(user.getId(),user.getUsername()+"测试");
        // 消息处理下游绑定事务，成功消费后提交ack
        // 手动提交offset
        ack.acknowledge();
        return new VkPublicResponse();
    }

    /**
     * 指定topic、partition、offset消费
     * @KafkaListener注解已全部为我们提供
     * topics和topicPartitions不能同时使用；
     * 监听TOPIC1的0号分区，同时监听TOPIC的0号分区和topic2的1号分区里面offset从3开始的消息
     */
//    @KafkaListener(id = "consumer1", containerGroup = "${spring.kafka.consumer.group-id2}",
//            topicPartitions = {
//            @TopicPartition(topic = KafkaConfig.TOPIC1, partitions = { "0" }),
//            @TopicPartition(topic = KafkaConfig.TOPIC, partitions = "0", partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "3"))
//    })
    public VkPublicResponse onMessage2(ConsumerRecord<?, ?> record){
        // 消费的哪个topic、partition的消息,打印出消息内容
        log.info("consumer1开始消费--->topic名字:"+record.topic()+",分区partition:"+record.partition()+",消费数据"+record.value());
        //解析收到的队列内容
        Object value = record.value();
        JSONObject jsonMsg = JSONObject.parseObject(value.toString());
        VkSmbmsUser user = createUssCrmEntity(jsonMsg);
        mqBasicsKafkaConService.updateUserById(user.getId(),user.getUsername()+"测试");
        return new VkPublicResponse();
    }

    /**
     * 消费者批量消费
     */
    //@KafkaListener(id = "consumer2", topics = KafkaConfig.TOPIC1 ,groupId = "felix-group")
    public VkPublicResponse onMessage3(List<ConsumerRecord<?, ?>> records) {
        log.info(">>>批量消费一次，records.size()="+records.size());
        for (ConsumerRecord<?, ?> record : records) {
            log.info((String) record.value());
            //解析收到的队列内容
            Object value = record.value();
            JSONObject jsonMsg = JSONObject.parseObject(value.toString());
            VkSmbmsUser user = createUssCrmEntity(jsonMsg);
            mqBasicsKafkaConService.updateUserById(user.getId(),user.getUsername()+"优秀");
        }
        return new VkPublicResponse();
    }

    /**
     * 消费者异常处理
     * 将异常处理器的BeanName放到@KafkaListener注解的errorHandler属性里面
     */
    //@KafkaListener(id = "consumer3", topics = KafkaConfig.TOPIC1, groupId = "${spring.kafka.consumer.group-id3}",errorHandler = "consumerAwareErrorHandler")
    public void onMessage4(ConsumerRecord<?, ?> record) throws Exception {
        // 消费的哪个topic、partition的消息,打印出消息内容
        log.info("consumer3开始消费--->topic名字:"+record.topic()+",分区partition:"+record.partition()+",消费数据"+record.value());
        throw new VkPublicException(VkErrorCodeEnum.MESSAGE_CONS_ERROR, "简单消费-模拟异常");
    }

    // 批量消费也一样，异常处理器的message.getPayload()也可以拿到各条消息的信息
    //@KafkaListener(id = "consumer4",topics = KafkaConfig.TOPIC1, groupId = "${spring.kafka.consumer.group-id3}", errorHandler="consumerAwareErrorHandler")
    public void onMessage5(List<ConsumerRecord<?, ?>> records) throws Exception {
        log.info("批量消费一次...");
        throw new VkPublicException(VkErrorCodeEnum.MESSAGE_CONS_ERROR, "批量消费-模拟异常");
    }

    /**
     * 消息过滤监听
     */
    //@KafkaListener(topics = KafkaConfig.TOPIC1, groupId = "${spring.kafka.consumer.group-id3}", containerFactory = "filterContainerFactory")
    public VkPublicResponse onMessage6(ConsumerRecord<?, ?> record) {
        log.info(String.valueOf(record.value()));
        return new VkPublicResponse();
    }


    /**
     * 消息转发
     * 从TOPIC1接收到的消息经过处理后转发到TOPIC
     */
    @KafkaListener(id = "consumer5", topics = KafkaConfig.TOPIC, groupId = "${spring.kafka.consumer.group-id3}")
    @SendTo(KafkaConfig.TOPIC1)
    public String onMessage7(ConsumerRecord<String, String> record) {
        log.info("先到:"+ KafkaConfig.TOPIC +"的消息经过处理的消息转发到:" + KafkaConfig.TOPIC1);
        return String.valueOf(record.value());
    }

    @KafkaListener(topics = {KafkaConfig.TOPIC1}, id = "consumer", containerGroup = "${spring.kafka.consumer.group-id1}")
    public VkPublicResponse onMessage8(ConsumerRecord<String, String> record){
        // 消费的哪个topic、partition的消息,打印出消息内容
        log.info("consumer开始消费--->topic名字:"+record.topic()+",分区partition:"+record.partition()+",消费数据"+record.value());
        //解析收到的队列内容
        String value = record.value();
        log.info("上一个topic转发过来的消息:" + value);
        JSONObject jsonMsg = JSONObject.parseObject(value);
        VkSmbmsUser user = createUssCrmEntity(jsonMsg);
        mqBasicsKafkaConService.updateUserById(user.getId(),user.getUsername()+"测试");

        return new VkPublicResponse();
    }


    // 监听器
    //@KafkaListener(id="timingConsumer", topics = KafkaConfig.TOPIC, groupId = "${spring.kafka.consumer.group-id3}", containerFactory = "delayContainerFactory")
    public void onMessage9(ConsumerRecord<?, ?> record){
        log.info("消费成功："+record.topic()+"-"+record.partition()+"-"+record.value());
    }

    /**
     * 队列消息解析处理
     * @param jsonMsg
     * @return
     */
    private VkSmbmsUser createUssCrmEntity(JSONObject jsonMsg) {
        VkSmbmsUser user = new VkSmbmsUser();
        user.setId(Integer.parseInt(jsonMsg.getString("id")));
        user.setUsercode(jsonMsg.getString("usercode"));
        user.setUsername(jsonMsg.getString("username"));
        user.setUserpassword(jsonMsg.getString("userpassword"));
        user.setGender(Integer.parseInt(jsonMsg.getString("gender")));
        user.setBirthday(jsonMsg.getString("birthday"));
        user.setPhone(jsonMsg.getString("phone"));
        user.setAddress(jsonMsg.getString("address"));
        user.setUserrole(Integer.parseInt(jsonMsg.getString("userrole")));
        return user;
    }

}